package com.lb.bs.client.zookeeper;

import com.lb.bs.client.factory.SingletonFactory;
import com.lb.bs.client.serialize.SerializationHolder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * program: LB-BS-Client
 * author: bsworld.xie
 * create: 2019-11-27 17:49
 * description:
 */
public class ZKMgr {
    private static final Logger logger = LoggerFactory.getLogger(ZKMgr.class);
    private CuratorFramework client;
    private static boolean initFlag;
    private static ZKMgr zkMgr;
    private static String zkString;
    private static String nameSpace;
    private static WatchMgr watchMgr;

    public static ZKMgr getInstance(String zkString, String nameSpace) {
        if (zkMgr == null) {
            synchronized (ZKMgr.class) {
                if (zkMgr == null) {
                    ZKMgr.zkString = zkString;
                    ZKMgr.nameSpace = nameSpace;
                    zkMgr = new ZKMgr(zkString, nameSpace);
                    watchMgr = SingletonFactory.getInstance(WatchMgr.class);
                    initFlag = true;
                }
            }
        }
        return zkMgr;
    }

    public static ZKMgr getInstance() {
        if (!initFlag) {
            throw new IllegalStateException("zkMgr has not been initial");
        }
        if (zkMgr == null) {
            synchronized (ZKMgr.class) {
                if (zkMgr == null) {
                    zkMgr = new ZKMgr(zkString, nameSpace);
                    initFlag = true;
                }
            }
        }
        return zkMgr;
    }

    private ZKMgr(String zkString, String nameSpace) {
        connect(zkString, nameSpace);
    }

    public void reconnect() {
        connect(zkString, nameSpace);
    }


    private void connect(String zkString, String nameSpace) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
//        ZookeeperFactory zookeeperFactory = new SalZookeeperFactory();
//        builder.zookeeperFactory(zookeeperFactory);
        CuratorFramework newClient = builder
                .connectString(zkString)
                .namespace(nameSpace)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .build();
        newClient.start();
        this.client = newClient;
        CountDownLatch countDownLatch = new CountDownLatch(1);
        client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                boolean connected = newState.isConnected();
                if (connected) {
                    countDownLatch.countDown();
                }
            }
        });

        try {
            countDownLatch.await(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            logger.info("connect(),countDownLatch.await error:", e);
        }
        logger.info("ZKMgr.class, connection, success, zkString:{},nameSpace:{}", zkString, nameSpace);
    }


    public void createPath(String path, String data) {
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, data.getBytes());
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    public void removePath(String path) {
        try {
            client.delete().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public String getData(String itemKey) {
        try {
            boolean exist = exist(itemKey);
            if (!exist) {
                return null;
            }
            byte[] bytes = client.getData().forPath(itemKey);
            if (bytes != null && bytes.length > 0) {
                return SerializationHolder.getHolder().deserialize(bytes);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void watch(String groupName) {
        boolean exist = exist(groupName);
        if (!exist) {
            return;
        }
        watchMgr.watch(groupName);
    }

    public boolean exist(String path) {
        try {
            Stat stat = client.checkExists().forPath(path);
            if (stat == null) {
                return false;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    public void close() {
        if (client != null) {
            client.close();
        }
    }


    CuratorFramework getClient() {
        return client;
    }


}
